package netty.mqtt.t1;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

public class NettyMqttClient {

    private final String host;
    private final int port;

    public NettyMqttClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(
                                    new LoggingHandler(LogLevel.INFO),
                                    MqttEncoder.INSTANCE,
                                    new MqttDecoder(),
                                    new MqttClientHandler()
                            );
                        }
                    });

            ChannelFuture future = bootstrap.connect(host, port).sync();
            System.out.println("Connected to MQTT server at " + host + ":" + port);
            future.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        String host = "localhost"; // MQTT 服务器地址
        int port = 1883;           // MQTT 默认端口
        new NettyMqttClient(host, port).start();
    }
}

class MqttClientHandler extends SimpleChannelInboundHandler<MqttMessage> {

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 发送 CONNECT 报文
        MqttConnectMessage connectMessage = MqttMessageBuilders.connect()
                .clientId("NettyMQTTClient")
                .keepAlive(60)
                .cleanSession(true)
                .build();
        ctx.writeAndFlush(connectMessage);
        System.out.println("Sent CONNECT message.");
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) {
        if (msg.fixedHeader() == null) {
            ctx.close();
            return;
        }

        switch (msg.fixedHeader().messageType()) {
            case CONNACK:
                System.out.println("Received CONNACK message.");
                sendSubscribe(ctx);
                break;

            case SUBACK:
                System.out.println("Received SUBACK message.");
                sendPublish(ctx);
                break;

            case PUBLISH:
                MqttPublishMessage publishMessage = (MqttPublishMessage) msg;
                String topic = publishMessage.variableHeader().topicName();
                String payload = publishMessage.payload().toString(io.netty.util.CharsetUtil.UTF_8);
                System.out.println("Received message from topic " + topic + ": " + payload);
                break;

            default:
                System.out.println("Unhandled message type: " + msg.fixedHeader().messageType());
        }
    }

    private void sendSubscribe(ChannelHandlerContext ctx) {
        // 发送 SUBSCRIBE 报文
        MqttSubscribeMessage subscribeMessage = MqttMessageBuilders.subscribe()
                .messageId(1)
                .addSubscription(MqttQoS.AT_LEAST_ONCE, "test/topic")
                .build();
        ctx.writeAndFlush(subscribeMessage);
        System.out.println("Sent SUBSCRIBE message.");
    }

    private void sendPublish(ChannelHandlerContext ctx) {
        // 发送 PUBLISH 报文
        MqttPublishMessage publishMessage = MqttMessageBuilders.publish()
                .topicName("test/topic")
                .qos(MqttQoS.AT_LEAST_ONCE)
                .messageId(344)
                .payload(ctx.alloc().buffer().writeBytes("Hello MQTT!".getBytes()))
                .build();
        ctx.writeAndFlush(publishMessage);
        System.out.println("Sent PUBLISH message.");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}
